package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.util.MyKafkaUtil;
import com.atguigu.edu.realtime.util.MySqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author ZXW
 * @version 1.0
 * @description 交易域，加购事实表
 * @date 2022/10/14 21:29
 */
public class DwdTradeCartAdd {
    public static void main(String[] args) {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //1.3 指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 2.检查点相关的设置(略)
        //TODO 3.从kafka的topic_db主题中读取数据 创建动态表
        String groupId = "dwd_trade_cart_add_group";
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL(groupId));
        // tableEnv.executeSql("select * from topic_db").print();

        //TODO 4.过滤出加购数据
        Table cartAdd = tableEnv.sqlQuery(
                "select \n" +
                        "  data['id'] id,\n" +
                        "  data['user_id'] user_id,\n" +
                        "  data['course_id'] course_id,\n" +
                        "  data['cart_price'] cart_price,\n" +
                        "  data['session_id'] session_id,\n" +
                        "  ts\n" +
                        "  from topic_db\n" +
                        "  where `table` = 'cart_info' and `type` = 'insert'"
        );

        tableEnv.executeSql(" create table dwd_trade_cart_add(\n" +
                "  id string," +
                "  user_id string," +
                "  course_id string," +
                "  cart_price string," +
                "  session_id string," +
                "  ts string," +
                "  primary key(id) not enforced" +
                " )" + MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_cart_add"));


        //TODO 5.将数据写入到kafka主题
        cartAdd.executeInsert("dwd_trade_cart_add");

    }

}
